package com.mall.order.mq;

import com.mall.order.dal.entitys.Order;
import com.mall.order.dal.persistence.OrderMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
@Component
@Slf4j
public class Consumer {

    @Autowired
    OrderMapper orderMapper;

    private final static String addr = "localhost:9876";

    DefaultMQPushConsumer consumer;
    // 收消息

    @PostConstruct
    public void init() {
        // 创建一个消息消费者对象
        consumer = new DefaultMQPushConsumer("consumer_group");

        // 注册到注册中心
        consumer.setNamesrvAddr(addr);

        // 订阅主题
        try {
            consumer.subscribe("checkOrder","*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        //设置一个消息监听器
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    MessageExt messageExt = msgs.get(0);

                    byte[] body = messageExt.getBody();
                    String orderId = new String(body);

                    Order order = orderMapper.selectByPrimaryKey(orderId);
                    Integer status = order.getStatus();
                    if(status==0){
                        order.setStatus(5);
                        orderMapper.updateByPrimaryKeySelective(order);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                }catch (Exception e) {

                    e.printStackTrace();
                    log.info("消费异常");
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            }
        );

        // 启动
        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
